package com.atguigu.flink.java.chapter_7;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/12 20:29
 */
public class Flink08_Chapter07_Aggregate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(2);
        env
          .socketTextStream("mac", 9999)
          .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
              @Override
              public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                  Arrays.stream(value.split("\\W+")).forEach(word -> out.collect(Tuple2.of(word, 1L)));
              }
          })
          .keyBy(t -> t.f0)
          .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
          .aggregate(new AggregateFunction<Tuple2<String, Long>, Long, Long>() {

              // 创建累加器: 初始化中间值
              @Override
              public Long createAccumulator() {
                  System.out.println("createAccumulator");
                  return 0L;
              }

              // 累加器操作
              @Override
              public Long add(Tuple2<String, Long> value, Long accumulator) {
                  System.out.println("add");
                  return accumulator + value.f1;
              }

              // 获取结果
              @Override
              public Long getResult(Long accumulator) {
                  System.out.println("getResult");
                  return accumulator;
              }

              // 累加器的合并: 只有会话窗口才会调用
              @Override
              public Long merge(Long a, Long b) {
                  System.out.println("merge");
                  return a + b;
              }
          }, new ProcessWindowFunction<Long, String, String, TimeWindow>() {
              @Override
              public void process(String key, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
                  System.out.println("process ..." + elements.spliterator().estimateSize());
                  out.collect("a");
              }
          })
          .print();


        env.execute();
    }
}
